#include "dbctrl.h"
#include "dbconfig.h"
#include "cmd/cmdupdatedatatodb.h"

bool CDBCtrl::s_bReloadConfig = false;
bool CDBCtrl::s_bExit = false;
CDBCtrl* CDBCtrl::m_pstMain = NULL ;

void CDBCtrl::sigusr1_handle(int iSigVal)
{
    s_bReloadConfig = true;
    signal(SIGUSR1, CDBCtrl::sigusr1_handle);
}

void CDBCtrl::sigusr2_handle(int iSigVal)
{
    s_bExit = true;
    signal(SIGUSR2, CDBCtrl::sigusr2_handle);
}

CDBCtrl::CDBCtrl()
{
    CTMySQLCharset stTemp; //临时增加，确保.a会被载入

    m_bResetShm = true;

    // 统计项描述
    const char g_aszEnpStatDesc[][CTStatItem::CTLIB_DESC_MAX] = 
    {
        "RecvMessage",
        "SendMessage",
        "Accept",
        "IdleClose",
        "MainLoop",
        "WaitTimeout",
        "Check",
    };

    m_stStat.Init(CT_COUNTOF(g_aszEnpStatDesc), g_aszEnpStatDesc);
}

CDBCtrl::~CDBCtrl()
{
}

int CDBCtrl::Init()
{
    int iRet = 0;
    m_pstMain = this;

    //安装信号处理函数
    signal(SIGUSR1, CDBCtrl::sigusr1_handle);	// kill -USR1 pid 会让程序重读配置
    signal(SIGUSR2, CDBCtrl::sigusr2_handle);	// kill -USR2 pid 会让程序安全的退出

    // 读取配置
    iRet = CONF->LoadConfig();
    CHECK_THROW(iRet);

    //设置数据库名称
    m_szDBName = CONF->szDBServer;

    // 创建Epoll
    iRet = m_stEpoll.Init(DB_EVENT_SIZE, DB_WAIT_SIZE);
    CHECK_THROW(iRet);

    iRet = m_stMpShmQueue.Init(false, 
        CONF->MpFrontEndShmKey(), CONF->MpFrontEndShmSize,
        CONF->MpFrontEndSocket(), CONF->MpBackEndSocket(),
        m_stEpoll,
        *this,
        &CDBCtrl::OnMpShmQueueEvent);
    CHECK_THROW(iRet);

    // 创建共享内存
#ifdef _DEBUG
    const char* pszConfig = "(512,3000)(1024,2000)(4096,1000)(16384,500)(65536,500)(300000,20)";
#elif defined _REG10000
    const char* pszConfig = "(512,2000)(1024,2000)(4096,2000)(16384,2000)(65536,2000)(131072,100)";
#else
    //const char* pszConfig = "(512,6000)(1024,6000)(4096,6000)(16384,6000)(65536,4000)(131072,100)";
    const char* pszConfig = "(512,2000)(1024,2000)(4096,2000)(16384,2000)(65536,2000)(131072,100)";
#endif

//////////////////////////////////////////////////////////////////////////update mgr 更新开始//////////////////////////////////////////////////////////////////////////
#ifdef _DEBUG
    const char* pszUpdateConfig = "(512,50)(1024,50)(4096,50)(16384,50)(32768,50)(65536,50)(300000,20)";
#elif defined _REG10000
    const char* pszUpdateConfig = "(512,320)(1024,3200)(2048,2000)(4096,2000)(16384,2000)(32768,4)(65536,2)";
#else
    //const char* pszUpdateConfig = "(512,500000)(1024,20000)(2048,50000)(4096,80000)(16384,5000)(32768,2000)(65536,200)";
    const char* pszUpdateConfig = "(512,320)(1024,3200)(2048,2000)(4096,2000)(16384,2000)(32768,4)(65536,2)";
#endif
    const size_t iUpdateMgrSize  = m_stUpdateMgr.CountSize(pszUpdateConfig);
    if (iUpdateMgrSize <= 0)
    {
        CT_ERROR(("invalid m_stUpdateMgr size %d", iUpdateMgrSize));
        return RET_SYS_ERROR;
    }

    const int iUptTotalSize = sizeof(int) + iUpdateMgrSize;

    iRet = m_stUpdateMgrShm.Create(CONF->UptShmKey(), iUptTotalSize);
    CHECK_RETURN(iRet);

    bool bResetUptShm = m_bResetShm;
    int* piOrgUptShmSize = (int*)m_stUpdateMgrShm.GetBuffer();
    if (!bResetUptShm && *piOrgUptShmSize != iUptTotalSize) //表示-u重启，需要恢复共享内存中的数据，但是共享内存中的长度和之前的不同
    {
        CT_WARNING(("upt shm size changed, -u param is invalid"));
        bResetUptShm = true;
    }
    *piOrgUptShmSize = iUptTotalSize;
    
    iRet = m_stUpdateMgr.Init(CDBCtrl::UpdateData, pszUpdateConfig, m_stUpdateMgrShm.GetBuffer() + sizeof(int),
        CONF->m_iInitFlushNum,CONF->m_iMaxFlushNum, bResetUptShm,true);
    CHECK_THROW(iRet);
    
//////////////////////////////////////////////////////////////////////////update mgr 更新结束//////////////////////////////////////////////////////////////////////////

    const size_t iFactorySize = CT_CMDFACTORY->CountSize(pszConfig);

    const size_t iUserTokenSize = m_stUserTokenToGID.CountSize(USER_TOKEN_TO_GID_COUNT);

    const size_t iCountSize = sizeof(int) + iFactorySize + iUserTokenSize;

//#ifndef WIN32
//    iRet = m_stMgrShm.Create(CTStringUtils::StrToInt<key_t>(CONF->MgrShmKey()), iCountSize);
//#else
    iRet = m_stMgrShm.Create(CONF->MgrShmKey(), iCountSize);
//#endif
    CHECK_THROW(iRet);

    size_t BufOffset = 0;

    int* piOrgShmSize = (int*)(m_stMgrShm.GetBuffer() + BufOffset);
    if (!m_bResetShm && (size_t)*piOrgShmSize != iCountSize) //表示-u重启，需要恢复共享内存中的数据，但是共享内存中的长度和之前的不同
    {
        CT_WARNING(("shm size changed, -u param is invalid, %d, %d", *piOrgShmSize, iCountSize));
        m_bResetShm = true;
    }
    *piOrgShmSize = iCountSize;
    BufOffset += sizeof(int);

    // 初始化异步命令工厂
    iRet = CT_CMDFACTORY->Init(pszConfig, m_stMgrShm.GetBuffer() + BufOffset, iFactorySize);
    CHECK_THROW(iRet);
    BufOffset += iFactorySize;

    iRet = m_stUserTokenToGID.Init(m_stMgrShm.GetBuffer() + BufOffset, USER_TOKEN_TO_GID_COUNT);
    CHECK_THROW(iRet);
    BufOffset += iUserTokenSize;

    m_stDecodeBuf.Attach(NULL,
        CTMsgHead::EMAX_MSGLEN + sizeof(CTMsgHead) + sizeof(CEnpNetHead), 0);

    m_stEncodeBuf.Attach(NULL,
        CTMsgHead::EMAX_MSGLEN + sizeof(CTMsgHead) + sizeof(CEnpNetHead), 0);

    // 最后创建ENP共享内存队列, 表示开始处理外部请求
    iRet = m_stEnpShmQueue.Init(true, 
        CONF->EnpFrontEndShmKey(), CONF->EnpFrontEndShmSize,
        CONF->EnpFrontEndSocket(), CONF->EnpBackEndSocket(),
        m_stEpoll,
        *this,
        &CDBCtrl::OnEnpShmQueueEvent);
    CHECK_THROW(iRet);

    CT_INFO(("dbsvr start"));
    return 0;
}

int CDBCtrl::Run()
{
    int iRet = 0;
    int iLoop = 0;

    m_uiNow = static_cast<unsigned int>(time(0));
    int iPrevNow = m_uiNow;

    while (true)
    {
        // 信号检查
        if (s_bReloadConfig)
        {
            s_bReloadConfig = false;
            CONF->ReloadConfig();
            m_stUpdateMgr.ReSetUpdateFlushNum(CONF->m_iInitFlushNum,CONF->m_iMaxFlushNum);
        }

        if (s_bExit)
        {
            s_bExit = false;
            CT_TRACE(("dbsvr end!"));
            return 0;
        }

        m_stStat.Put(db_stat_mainloop);

        // 定时器检查
        if (iLoop == 0)
        {
            m_uiNow = (unsigned int) time(0);
            if (m_uiNow - iPrevNow >= 15)
            {
                CT_CMDFACTORY->CheckTimeoutCmd(m_uiNow);
                iPrevNow = m_uiNow;
                DumpStatInfo();
            }
        }

        //提高刷新检查的频率，使数据刷新更平滑
        time_t iNowTime = time(NULL);
        m_stUpdateMgr.CheckUpdate(3,iNowTime);

        iRet = m_stEpoll.WaitAndEvent(10);
        if (iRet == 0) // 超时了
        {
            m_stStat.Put(db_stat_waittimeout);
        }

        // 处理异步命令队列中命令
        iRet = CT_CMDFACTORY->DoQueue();
        CHECK_THROW(iRet);

        HandleEnpMsg();
        HandleMpMsg();

        if (++iLoop >= 50)
        {
            iLoop = 0;
            m_stStat.Put(db_stat_check);

            int iTime = m_stUpdateMgr.GetLastSwapTime();

            //当hash满或者时间太长，系统强制检查swap
            if(m_stUpdateMgr.CheckPassUsedPentage(100) || (iNowTime - iTime) > 600)
            {
                bool bIsSwap = false;
                m_stUpdateMgr.CheckSwap(10,iNowTime,50,&bIsSwap);
                if (bIsSwap)
                {
                    CT_UPDATE(("------------------system-force-to-swap--------------------"));
                }
            }       
        }
    }

    return 0;
}

int CDBCtrl::Exit()
{
	m_stUpdateMgr.ReSetUpdateFlushNum(CONF->m_iInitStopFlushNum,CONF->m_iMaxStopFlushNum);
	CT_INFO(("before exit update flush num %d %d",CONF->m_iInitStopFlushNum, CONF->m_iMaxStopFlushNum));

    while(!m_stUpdateMgr.CheckEmpty())
    {
        int iNowTime = time(NULL);
        //只有数据刷完才停机
        m_stUpdateMgr.CheckUpdate(2,iNowTime);
        bool bSwapMgr = false;
        m_stUpdateMgr.CheckSwap(2,iNowTime,20, &bSwapMgr);
        if (bSwapMgr)
        {
            CT_UPDATE(("--------------------swap--------------------"));
        }
        m_stEpoll.WaitAndEvent(1000);
    }

    DumpStatInfo();
    CT_INFO(("%s exit", APPNAME));
    return 0;
}

void CDBCtrl::OnEnpShmQueueEvent(CDBEpollObject* pstObject, SOCKET iSocket, int iEvent)
{
    CT_TRACE(("OnEnpShmQueueEvent socket=%d event=%d", iSocket, iEvent));
    m_stEnpShmQueue.ClearNotify(); // 接收Notify数据, 否则会一直收到事件
    //TODO: 每个消息都要执行ClearNotify, 可能存在性能问题, 以后验证

    if (!CTEpoll::IsInputEvent(iEvent))
    {
        CT_WARNING(("ShmQueue recv not match event(%d) socket=%d", iEvent, iSocket));
        return;
    }

    HandleEnpMsg();
}


void CDBCtrl::OnMpShmQueueEvent(CDBEpollObject* pstObject, SOCKET iSocket, int iEvent)
{
    CT_TRACE(("OnMpShmQueueEvent socket=%d event=%d", iSocket, iEvent));
    m_stMpShmQueue.ClearNotify(); // 接收Notify数据, 否则会一直收到事件
    //TODO: 每个消息都要执行ClearNotify, 可能存在性能问题, 以后验证

    if (!CTEpoll::IsInputEvent(iEvent))
    {
        CT_WARNING(("ShmQueue recv not match event(%d) socket=%d", iEvent, iSocket));
        return;
    }

    HandleMpMsg();
}

void CDBCtrl::DumpStatInfo()
{
    m_stStat.Dump(CT_STAT);

    //TODO: 还要加上其他的统计信息
	CT_CMDFACTORY->DumpStatInfo(CT_STAT);
	CT_CMDFACTORY->DumpStatDetail(CT_STAT);

    //更新管理器
    CT_STAT->Log(EInfo, "%s",m_stUpdateMgr.GetStatInfo());
    
}

void CDBCtrl::HandleEnpMsg()
{
    int iRet = 0;
    CTCodeStream s;
    CTMsgHead stMsgHead;

    for (int i = 0; i < 50; i++)  //此处的50为性能测试时通过反复调整获得的值
    {
        if (!m_stEnpShmQueue.HasCode())
        {
            return;
        }

        // 接收消息
        int iCodeLen = 0;
        m_stDecodeBuf.Clear();
        iRet = m_stEnpShmQueue.GetOneCode(m_stDecodeBuf.GetFreeBuf(), m_stDecodeBuf.GetFreeLen(), iCodeLen);
        if (iRet || iCodeLen < (int)sizeof(CEnpNetHead))
        {
            CT_WARNING(("get code from enpshmqueue failed ret=%d, codelen=%d", iRet, iCodeLen));
            continue;
        }
        m_stDecodeBuf.Append(iCodeLen);

        // 先获取NetHead
        CEnpNetHead& stHead = *(CEnpNetHead*)m_stDecodeBuf.GetUsedBuf();
        if (iCodeLen != (int)sizeof(CEnpNetHead) + stHead.unDataLength) // 长度不一致
        {
            CT_WARNING(("code length invalid. %d != %d + %d", iCodeLen,
                sizeof(CEnpNetHead), stHead.unDataLength));
            continue;
        }

        if (stHead.unDataLength <= 0)
        {
			CT_WARNING(("stHead.unDataLength <= 0"));
            continue;
        }

        CT_TRACE(("get one msg from enp, datalen(msghead+body) = %d", stHead.unDataLength));

        //TODO: 考虑增加IP限制的功能(原型阶段不实现)

        // 这里都是处理来自client的请求
        iRet = s.Attach(m_stDecodeBuf.GetUsedBuf() + sizeof(CEnpNetHead), stHead.unDataLength);
        CHECK_THROW(iRet);

        s.InitConvert();
        iRet = CodeConvert(s, stMsgHead, NULL, bin_decode());

        if (iRet != 0)
        {
            CT_ERROR(("Error in decoding message from enp, ret = %d", iRet));
            continue;
        }
        //CHECK_THROW(iRet);

        if (!stMsgHead.IsValid())
        {
            CT_WARNING(("invalid msghead, msgid = %d, msgtype = %d", stMsgHead.shMsgID, stMsgHead.shMsgType));
            continue;
        }

        // 创建异步命令
        CTAsyncCmd* pstCmd = (CTAsyncCmd*) CT_CMDFACTORY->CreateCmd(stMsgHead.shMsgID);
        if (!pstCmd)
        {
            CT_WARNING(("create cmd %d return null", stMsgHead.shMsgID));
            continue;
        }

        // 异步命令执行完后能自己释放
        CCmdDoParam stParam((void*)this, stHead, stMsgHead, s);
        iRet = pstCmd->Do(&stParam);
        CT_TRACE(("do cmd over, ret %d", iRet));
    }
}

void CDBCtrl::HandleMpMsg()
{
    int iRet = 0;
    CTCodeStream s;
    CTMsgHead stMsgHead;

    for (;;)
    {
        if (!m_stMpShmQueue.HasCode())
        {
            return;
        }

        // 接收消息
        int iCodeLen = 0;
        m_stDecodeBuf.Clear();
        iRet = m_stMpShmQueue.GetOneCode(m_stDecodeBuf.GetFreeBuf(), m_stDecodeBuf.GetFreeLen(), iCodeLen);
        if (iRet || iCodeLen < 16) // CTMsgHead的编码长度是16
        {
            CT_WARNING(("get code from mpshmqueue failed ret=%d, codelen=%d", iRet, iCodeLen));            
            continue;
        }
        m_stDecodeBuf.Append(iCodeLen);

		// 先获取NetHead
		CEnpNetHead& stHead = *(CEnpNetHead*)m_stDecodeBuf.GetUsedBuf();
		if (iCodeLen != (int)sizeof(CEnpNetHead) + stHead.unDataLength) // 长度不一致
		{
			CT_WARNING(("code length invalid. %d != %d + %d", iCodeLen,
				sizeof(CEnpNetHead), stHead.unDataLength));
			continue;
		}

		if (stHead.unDataLength <= 0)
		{
			CT_WARNING(("stHead.unDataLength <= 0"));
			continue;
		}

		CT_TRACE(("get one msg from mp, datalen(msghead+body) = %d", stHead.unDataLength));

        // 这里都是处理来自mpsvr的请求
        iRet = s.Attach(m_stDecodeBuf.GetUsedBuf() + sizeof(CEnpNetHead), 
							(unsigned short)(m_stDecodeBuf.GetUsedLen() & 0xFFFF));
        CHECK_THROW(iRet);
		
        s.InitConvert();
        iRet = CodeConvert(s, stMsgHead, NULL, bin_decode());
        CHECK_THROW(iRet);

        if (!stMsgHead.IsValid())
        {
            CT_WARNING(("msghead invalid, stMsgHead.shMsgID = %d, stMsgHead.shMsgType = %d", 
							stMsgHead.shMsgID, stMsgHead.shMsgType));
            continue;
        }

        // 根据ACT找到对应发出请求的异步命令
        CTAsyncCmdInf* pstCmd = reinterpret_cast<CTAsyncCmdInf*>(stMsgHead.llMsgAct);
        if (!CT_CMDFACTORY->IsValidCmd(pstCmd))
        {
            CT_WARNING((""));
            continue;
        }

        // 检查消息的Seq是否匹配
        if (pstCmd->GetCmdSeq() != stMsgHead.iCmdSeq)
        {
            CT_WARNING((""));
            continue;
        }

        // 执行响应函数
        CCmdOnAnswerParam stParam((void*)this, stMsgHead, s);
        iRet = pstCmd->OnAnswer(&stParam);
    }
}

int CDBCtrl::AnswerToClient(CEnpNetHead& stHead, CTCodeStream& s)
{
    CT_TRACE((""));
    int iRet = 0;
    stHead.unDataLength = s.m_ushPos;

    iRet = m_stEnpShmQueue.PutOneCode((const char*)&stHead, sizeof(stHead), s.m_pszBuf, s.m_ushPos);
    if (iRet)
    {
        CT_WARNING(("ret = %d", iRet));
        return iRet;
    }

	CT_TRACE(("send %d bytes to client", stHead.unDataLength));

    return iRet;
}

int CDBCtrl::AnswerToClient(CEnpNetHead& stHead, CTMsgHead& stMsgHead, CResult& stRet)
{
    CTCodeStream s(m_stEncodeBuf);
    CodeConvert(s, stMsgHead, NULL, bin_encode());
    CodeConvert(s, stRet, NULL, bin_encode());
    if (s.m_iRet)
    {
        CT_WARNING((""));
        return s.m_iRet;
    }

    return AnswerToClient(stHead, s);
}

int CDBCtrl::AnswerToClient(CEnpNetHead& stHead, const char* sAnsMsg, int MsgLen)
{
    int iRet = 0;
    stHead.unDataLength = MsgLen & 0xFFFF;
    iRet = m_stEnpShmQueue.PutOneCode((const char*)&stHead, sizeof(stHead), sAnsMsg, MsgLen);
    if (iRet)
    {
        CT_WARNING((""));
        return iRet;
    }

    return iRet;
}

int CDBCtrl::RequestToMp(const char* pszBuf, int iBufLen)
{
    return m_stMpShmQueue.PutOneCode(pszBuf, iBufLen);
}

int CDBCtrl::UpdateData(CVarObject * pstObj,CVarObjectKey * pstKey)
{
    if(pstObj == NULL || pstKey == NULL)
    {
        return -1;
    }

    CCmdUpdateDataToDB * pstCmd = (CCmdUpdateDataToDB*)CT_CMDFACTORY->CreateCmd(ID_DB_INNER_UPDATEDATA);
    if (!pstCmd)
    {
        return -1;
    }

    pstCmd->m_pstDB = CDBCtrl::m_pstMain;
    pstCmd->m_stReq.pszData    = pstObj->m_chData;
    pstCmd->m_stReq.iDataLen   = pstObj->m_stDataHead.m_iDataLen;
    pstCmd->m_stReq.m_iOpType  = pstObj->m_stDataHead.m_iDataInfo;
    pstCmd->m_stReq.m_stKey    = *pstKey;
    pstCmd->m_stReq.m_uiSwapCount = CDBCtrl::m_pstMain->m_stUpdateMgr.GetSwapCount();

    int iRet = pstCmd->Do(NULL);
    if (iRet != 0)
    {
        CT_WARNING(("CCmdUpdateData do() error"));
        return -1;
    }
    CT_TRACE((""));
    return 0;
}


